package com.demo.sse.service;

import com.demo.sse.model.ServerMessage;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Random;

@Service
public class SseService {
    private final Sinks.Many<ServerMessage> sink;
    private final AtomicInteger[] counters;
    private static final int COUNTER_COUNT = 5;  // 计数器数量
    private final Random random = new Random();  // 用于生成随机数

    public SseService() {
        this.sink = Sinks.many().multicast().onBackpressureBuffer();
        this.counters = new AtomicInteger[COUNTER_COUNT];
        // 初始化计数器，设置随机初始值
        for (int i = 0; i < COUNTER_COUNT; i++) {
            int initialValue = random.nextInt(100); // 0-99的随机初始值
            counters[i] = new AtomicInteger(initialValue);
        }
    }

    @PostConstruct
    public void init() {
        // 为每个计数器创建定时任务，并错开启动时间
        for (int i = 0; i < COUNTER_COUNT; i++) {
            final int counterId = i;
            // 每个计数器的定时任务延迟启动时间不同，避免同时触发
            int initialDelay = i * 400; // 每个计数器错开400毫秒启动
            
            Flux.interval(Duration.ofMillis(initialDelay), Duration.ofSeconds(2))
                .subscribe(tick -> {
                    int currentValue = counters[counterId].incrementAndGet();
                    ServerMessage message = new ServerMessage(
                        UUID.randomUUID().toString(),
                        String.valueOf(currentValue),
                        System.currentTimeMillis(),
                        counterId
                    );
                    sink.tryEmitNext(message);
                });
        }
    }

    public Flux<ServerMessage> getMessageStream() {
        return sink.asFlux();
    }
} 